package com.example.mqttmodule.config;

import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.net.ssl.SSLContext;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;

@Configuration
@EnableIntegration
@Log4j2
public class MqttConfig {

    @Value("${mqtt.broker-url}")
    private String brokerUrl;

    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    @Value("${mqtt.client-id}")
    private String clientId;

    @Value("${mqtt.topics.status}")
    private String topic;

    @Value("${mqtt.completionTimeout}")
    private int completionTimeout;

    @Value("${mqtt.keepalive}")
    private int keepalive;

    @Value("${mqtt.qos}")
    private int qos;

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{brokerUrl});
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
        // 这里设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(false);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(completionTimeout);
        options.setExecutorServiceTimeout(completionTimeout);
        options.setKeepAliveInterval(keepalive);
        options.setAutomaticReconnect(true);

        //SSL/TLS brokerUrl: ssl://localhost:8883
        //不开启  brokerUrl: tcp://broker.emqx.io:1883
//        try {
//            SSLContext sslContext = SSLContext.getInstance("TLS");
//            sslContext.init(null, null, null);
//            options.setSocketFactory(sslContext.getSocketFactory());
//        } catch (Exception e) {
//            log.info("SSL/TLS init error");
//            throw new RuntimeException(e);
//        }

        //遗嘱消息
        options.setWill("will/topic", "connection lost".getBytes(), 2, true);
//        client.setCallback(new MqttPushCallback());
        return options;
    }

    @Bean
    public MqttPushCallback mqttPushCallback() {
        return new MqttPushCallback();
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory(MqttPushCallback callback) {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(){
            @Override
            public IMqttAsyncClient getAsyncClientInstance(String serverURI, String clientId) throws MqttException {
                IMqttAsyncClient client = super.getAsyncClientInstance(serverURI, clientId);
                client.setCallback(callback);  // 设置回调
                return client;
            }
        };

        factory.setConnectionOptions(mqttConnectOptions());
        return factory;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound(MqttPahoClientFactory factory) {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(clientId + "-publisher", factory);
        System.out.println("MqttPahoMessageHandler: " + handler.getClientId());
        handler.setAsync(true);
        handler.setDefaultTopic(topic);
        handler.setDefaultQos(qos); // 建议明确设置QoS
        return handler;
    }
}
